package com.liyueheng.app.dataWarehouseDetail.rfm

import com.liyueheng.util.{ConfigLoader, SaveAsTable, SparkConf}
import org.apache.spark.sql.functions._

object MonetaryScore {
  def calcM(): Unit = {
    println("------------------ 计算消费金额分数 -----------------")
    val spark = SparkConf.createSparkSession("RFM-M")
    val dwd = ConfigLoader.getString("databases.dwd")
    val dws = ConfigLoader.getString("databases.dws")

    val userActDF = spark.table(s"$dwd.user_act").filter("act_type = 2")
    val skuDF = spark.table(s"$dwd.sku")

    val result = userActDF.join(skuDF, userActDF("sku") === skuDF("sku_id"))
      .filter(month(col("act_date")).between(2, 4))
      .groupBy("user")
      .agg(round(sum("price"), 2).alias("monetary"))

    SaveAsTable.saveAsTable(result, s"$dws.detail_rfm_monetary")
    SparkConf.stopSparkSession(spark)
  }
}
